package com.study.config;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author mjy(装配RocketMq监听)
 * @date 2021/12/16
 * @description
 */
@Configuration
public abstract class RocketMqConsumerConfiguration {

    Logger log = LoggerFactory.getLogger(RocketMqConsumerConfiguration.class);

    @Autowired
    private RocketMqConsumerConfig rocketMqConsumerConfig;

    @Bean
    public DefaultMQPushConsumer createConsumer() throws MQClientException {
        log.info("------开始创建rocketMq消息者----");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMqConsumerConfig.getGroupName());
        //指定nameServer连接地址
        consumer.setNamesrvAddr(rocketMqConsumerConfig.getNameServer());
        //指定订阅主题以及tag
        consumer.subscribe(rocketMqConsumerConfig.getDeviceTopic(), rocketMqConsumerConfig.getDeviceTag());
        log.info("nameServer={},deviceTopic={},deviceTag={}", rocketMqConsumerConfig.getNameServer(), rocketMqConsumerConfig.getDeviceTopic(), rocketMqConsumerConfig.getDeviceTag());
        //创建rocketMq监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
             return RocketMqConsumerConfiguration.this.dealMessage(msgs);
            }
        });
        consumer.start();
        return consumer;
    }


    /**
     * 处理消息
     *
     * @param msgs
     * @return
     */
    public abstract ConsumeConcurrentlyStatus dealMessage(List<MessageExt> msgs);
}
